package cn.com.bluemoon.daps.sync.service.impl;

import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.BmStatus;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapException;
import cn.com.bluemoon.daps.common.exception.DapThrowException;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.compare.ModelCompareClient;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTask;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTaskDetail;
import cn.com.bluemoon.daps.sync.service.*;
import cn.com.bluemoon.daps.sync.vo.applypublish.ApplyPublishParam;
import cn.com.bluemoon.daps.sync.vo.applypublish.ApplyPublishResultVo;
import cn.com.bluemoon.daps.sync.vo.applypublish.ModelCompareVo;
import cn.com.bluemoon.daps.sync.vo.applypublish.TableApplyPublishReportVo;
import cn.com.bluemoon.daps.sync.vo.model2prod.SyncProTaskDto;
import cn.hutool.db.Db;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import javax.annotation.Resource;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static cn.com.bluemoon.daps.sync.base.BmScript.PT_SQL_SPLIT_CHAR;
import static cn.com.bluemoon.daps.sync.base.BmScript.SQL_SPLIT_CHAR;

/**
 * 申请发布
 * 1.负责提供对外接口：模型版本对比
 * 2.拉取模型版本数据
 * 3.使用同步公共对比库，生成对比结果
 * 4.入库到发布同步任务列表中（禁用状态数据，暂不提供进行发布同步）
 * 5.返回给模型-申请发布调用者
 * 6.发布通过，触发同步任务状态调整为启用
 * 7.发布撤销，触发同步任务状态为删除
 * 8.进入发布同步列表
 *
 * @author Jarod.Kong
 */
@Slf4j
@Service
public class DapModelApplyPublishServiceImpl implements DapModelApplyPublishService {
    private static final ExecutorService layerCheckExecutor = Executors.newWorkStealingPool(80);

    @Resource
    private DapSyncBasicService syncBasicService;
    @Resource
    private DapSyncProdEnvTaskService taskService;
    @Resource
    private DapSyncProdEnvTaskDetailService taskDetailService;
    @Resource
    private DapSyncTaskOperRecordService taskOperRecordService;

    private static String concatAndLine(String str, String dom) {
        if (str == null) {
            return "";
        }
        return str.concat(dom);
    }

    /**
     * 版本修复：针对表重建+修改/删除字段的操作进行版本关系修复
     *
     * @param modelV1 v1 old last
     * @param modelV2 v2 newest
     * @return 被修复的表名
     */
    private static Set<String> versionRepair(List<TableModel> modelV1, List<TableModel> modelV2) {
        Set<String> repairTableNames = new HashSet<>();
        Map<String, TableModel> engNameAndTableModel = modelV2.stream().collect(Collectors.toMap(t -> t.getTable().getEngName(), t -> t, (a, b) -> b));
        for (TableModel tm1 : modelV1) {
            String engName1 = tm1.getTable().getEngName();
            TableModel tm2 = engNameAndTableModel.get(engName1);
            if (tm2 != null && !StringUtils.equals(tm1.getTable().getPId(), tm2.getTable().getId())) {
                repairTableNames.add(engName1);
                tm2.getTable().setBmKey(tm1.getTable().getPId());
                // 修正字段、分区、索引
                Map<String, DapDataModelTableField> fieldNameAndFields = tm1.getFieldNameAndFields();
                for (DapDataModelTableField f2 : tm2.getFields()) {
                    DapDataModelTableField f1 = fieldNameAndFields.get(f2.getFieldName());
                    if (f1 != null && StringUtils.equals(f2.getFieldName(), f1.getFieldName()) && !StringUtils.equals(f2.getId(), f1.getPId())) {
                        f2.setBmKey(f1.getPId());
                    }
                }
                Map<String, DapDataModelSubordinateKey> indexNameAndIndexes = tm1.getIndexNameAndIndexes();
                for (DapDataModelSubordinateKey f2 : tm2.getIndexes()) {
                    DapDataModelSubordinateKey f1 = indexNameAndIndexes.get(f2.getKeyName());
                    if (f1 != null && StringUtils.equals(f2.getKeyName(), f1.getKeyName()) && !StringUtils.equals(f2.getId(), f1.getPId())) {
                        f2.setBmKey(f1.getPId());
                    }
                }
                Map<String, DapDataModelTablePartition> partitionFieldNameAndPartitions = tm1.getPartitionFieldNameAndPartitions();
                for (DapDataModelTablePartition f2 : tm2.getPartitions()) {
                    DapDataModelTablePartition f1 = partitionFieldNameAndPartitions.get(f2.getPartitionFieldName());
                    if (f1 != null && StringUtils.equals(f2.getPartitionFieldName(), f1.getPartitionFieldName()) && !StringUtils.equals(f2.getId(), f1.getPId())) {
                        f2.setBmKey(f1.getPId());
                    }
                }
            }
        }
        return repairTableNames;
    }

    /**
     * 标记对比结果中哪些表被修复了
     *
     * @param repairTableNames 修复表
     * @param r2               对比结果集
     */
    private static void markVersionTableRepaired(Set<String> repairTableNames, List<ModelCompareVo> r2) {
        r2.stream().filter(mc -> mc.getFieldCompares() != null && !mc.getFieldCompares().isEmpty())
                .filter(mc -> repairTableNames.contains(mc.getTableName()) && mc.getFieldCompares().stream().anyMatch(c -> c.getModifyType().equals(ModifyType.DROP)))
                .forEach(mc -> {
                    log.warn("发现存在字段重建的情况！！info:\n {}", mc);
                    mc.setRebuildTableAndDropField(true);
                });
    }

    /**
     * 对比前校验接口
     *
     * @param modelId       模型id
     * @param newVsOld      新vs旧 ？
     * @param forceOverride 是否强制拉取对比且覆盖上一次对比结果，true是 false否，默认为false
     * @param userInfo      申请人
     * @return 用户
     */
    @Override
    public ApplyPublishResultVo modelTableLayerPredicate(String modelId,
                                                         Boolean newVsOld, Boolean forceOverride,
                                                         UserInfoHolder.UserInfo userInfo) {
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, SyncToEnv.NONE, userInfo);
        // main
        List<TableModel> modelV2 = getLocalTableModel(globalConfig, BmKeyType.USED_ID, true, false);
        List<TableApplyPublishReportVo> reportVos = predicateTableLayer(modelId, globalConfig, modelV2);
        if (!reportVos.isEmpty()) {
            log.warn("发现模型表建表语句存在语法异常，{}", reportVos);
            return ApplyPublishResultVo.errorReport(reportVos);
        }
        return ApplyPublishResultVo.success();
    }

    /**
     * 对比最新模型数据与上一个版本数据的差异数据
     * 用于入库到发布同步中
     *
     * @param modelId       模型id
     * @param newVsOld      新vs旧 ？
     * @param forceOverride 是否强制拉取对比且覆盖上一次对比结果，true是 false否，默认为false
     * @param userInfo      申请人
     * @return 用户
     */
    @Override
    public ApplyPublishResultVo compareModelVersion(String modelId,
                                                    boolean newVsOld,
                                                    Boolean forceOverride,
                                                    UserInfoHolder.UserInfo userInfo) {
        final String account = userInfo.getAccount();
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, SyncToEnv.NONE, userInfo);
        // 判断模型任务单号是否存在非已同步的，若有报错不允许对比
        List<DapSyncProdEnvTask> notDoneTasks = getTaskByModelIdAndPredicate(modelId);
        this.forceOverrideTasks(forceOverride, notDoneTasks);
        BmAssetUtils.isTrue(notDoneTasks.isEmpty(), "当前模型发布队列存在未同步生产库数据，请先执行同步");
        // main
        Tuple2<List<TableModel>, List<TableModel>> m = getV12ModelByIdOrPid(globalConfig, false);
        List<TableModel> modelV1 = m.getT1();// old last
        List<TableModel> modelV2 = m.getT2();// newest
        // 版本repair
        Set<String> repairTableNames = versionRepair(modelV1, modelV2);
        List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR;
        globalConfig.setCanDropDbTable(true);
        globalConfig.setCheckSqlLayer(true);
        if (newVsOld) {
            compareR = ModelCompareClient.doCompare(ModelCompareClient.BizType.SQL, globalConfig, modelV1, modelV2);
        } else {
            compareR = ModelCompareClient.doCompare(ModelCompareClient.BizType.SQL, globalConfig, modelV2, modelV1);
        }
        log.info("两个版本对比结果：{}", compareR.size());
        if (compareR.isEmpty()) {
            return ApplyPublishResultVo.errorCompareResultEmpty();
        }
        List<ModelCompareVo> r2 = ModelCompareClient.compareResultConvertVo(compareR);
        log.debug("对比结果数据：{}", r2);
        // r2中存在表tableRepair
        markVersionTableRepaired(repairTableNames, r2);
        log.info("对比结果存在，返回前记录对比结果，提供结果版本id，进行后续调整任务状态");
        SyncProTaskDto syncProTaskDto = asyncSaveTaskAndDetails(globalConfig, compareR);
        DapSyncProdEnvTask task = syncProTaskDto.getTask();
        // 异步线程需要独立赋值用户id
        task.setCreateBy(account);
        boolean status = taskService.save(task);
        boolean saveBatch = taskDetailService.bmSaveBatch(syncProTaskDto.getTaskDetails());
        final String compareVersion = task.getId();
        log.info("入库任务明细：任务为taskId = {}, detailSize={}, status={}-{}", compareVersion, syncProTaskDto.getTaskDetails().size(), status, saveBatch);
        taskOperRecordService.mark(modelId, compareVersion, String.format("申请发布开始，生成模型发布数据，执行人：%s", userInfo.getUserName()));
        return ApplyPublishResultVo.applyPublish(compareVersion, r2);
    }

    /**
     * 判断表是否存在语法问题，
     *
     * @param modelId      模型
     * @param globalConfig 配置
     * @param modelV2      最新数据
     * @return 语法错误集 为空合法，不为空有问题
     */
    private List<TableApplyPublishReportVo> predicateTableLayer(String modelId, GlobalConfig globalConfig, List<TableModel> modelV2) {
        List<TableApplyPublishReportVo> reportVos = new ArrayList<>();
        List<TableApplyPublishReportVo> sqlReportVos = getTableCreateSqlReportVos(modelId, globalConfig, modelV2);
        reportVos = sqlReportVos.stream().filter(r -> Boolean.FALSE.equals(r.getCanExec())).collect(Collectors.toList());
        return reportVos;
    }

    /**
     * 获取模型表数据的建表语句是否可执行
     *
     * @param modelId      模型
     * @param globalConfig 配置
     * @param modelV2      数据
     * @return 报告
     */
    private List<TableApplyPublishReportVo> getTableCreateSqlReportVos(String modelId, GlobalConfig globalConfig, List<TableModel> modelV2) {
        final Db sandboxDb;
        try {
            boolean status = globalConfig.rebuildDbName(true);
            log.info("申请发布前重建库名-清空所有表, status:{}", status);
            sandboxDb = globalConfig.findSandboxDb();
        } catch (Exception e) {
            throw new DapThrowException("申请发布-语法检查-创建连接信息失败", e);
        }
        final Db finalSandboxDb = sandboxDb;
        return modelV2.parallelStream().map(t -> {
            Future<TableApplyPublishReportVo> future = layerCheckExecutor.submit(() -> {
                DapDataModelTable table = t.getTable();
                List<String> createTableSql = t.getCreateTableSqls();
                String msg = null;
                if (createTableSql != null) {
                    msg = execCreateTableSql(finalSandboxDb, table, createTableSql);
                } else {
                    msg = "生成表" + table.getEngName() + "建表语句失败";
                }
                TableApplyPublishReportVo reportVo = new TableApplyPublishReportVo();
                reportVo.setModelId(modelId);
                reportVo.setTableId(table.getId());
                reportVo.setTableName(table.getEngName());
                reportVo.setCanExec(msg == null);
                reportVo.setMsg(msg);
                reportVo.setCreateTableSql(t.getCreateTableSqlPretty());
                reportVo.setType(TableApplyPublishReportVo.Type.SQL_VALID);
                return reportVo;
            });
            TableApplyPublishReportVo reportVo = null;
            try {
                reportVo = future.get();
            } catch (InterruptedException | ExecutionException e) {
                log.error("异步执行失败");
                reportVo = new TableApplyPublishReportVo();
                reportVo.setModelId(modelId);
                DapDataModelTable table = t.getTable();
                reportVo.setTableId(table.getId());
                reportVo.setTableName(table.getEngName());
                reportVo.setCanExec(false);
                reportVo.setMsg("异步执行建表语句失败");
                reportVo.setCreateTableSql(t.getCreateTableSqlPretty());
                reportVo.setType(TableApplyPublishReportVo.Type.SQL_VALID);
            }
            return reportVo;
        }).collect(Collectors.toList());
    }

    private String execCreateTableSql(Db sandboxDb, DapDataModelTable table, List<String> createTableSql) {
        String msg;
        List<String> execMsg = createTableSql.stream().filter(StringUtils::isNotBlank).map(sql -> {
            try {
                sandboxDb.execute(sql);
                log.debug("执行表{}建表语句{}，成功", table.getEngName(), sql);
            } catch (SQLException e) {
                log.warn("申请发布检查表{}建表语句{}，执行失败", table.getEngName(), sql, e);
                return e.getMessage();
            }
            return "";
        }).filter(StringUtils::isNotBlank).collect(Collectors.toList());
        log.info("执行表{}建表语句{}", table.getEngName(), createTableSql);
        msg = execMsg.isEmpty() ? null : String.join(";\n", execMsg);
        return msg;
    }

    private List<DapSyncProdEnvTask> getTaskByModelIdAndPredicate(String modelId) {
        LambdaQueryWrapper<DapSyncProdEnvTask> neDoneWrapper = Wrappers.<DapSyncProdEnvTask>lambdaQuery()
                .in(DapSyncProdEnvTask::getBmStatus, BmStatus.DISABLE, BmStatus.ON)
                .eq(DapSyncProdEnvTask::getModelId, modelId)
                .and(w -> w.ne(DapSyncProdEnvTask::getPreProdStatus, SyncStatus.SYNC_DONE)
                        .or()
                        .ne(DapSyncProdEnvTask::getProdStatus, SyncStatus.SYNC_DONE)
                );
        List<DapSyncProdEnvTask> notDoneTasks = taskService.list(neDoneWrapper);
        return notDoneTasks;
    }

    /**
     * 用r2，其中存在删表a，创建表a的用r1中的修改表来替换
     *
     * @param r1 根据表名对比、字段id、索引id、分区id对比
     * @param r2 根据表id、字段id、索引id、分区id对比
     * @param g  全局
     * @return 结果
     */
    private List<ModelCompareVo> mergeModelCompare(List<ModelCompareVo> r1, List<ModelCompareVo> r2, GlobalConfig g) {
        if (r1 == null || r1.isEmpty() || r2 == null || r2.isEmpty()) {
            return Collections.emptyList();
        }
        // r2是否存在删除表的操作
        List<ModelCompareVo> dropTableCrs = r2.stream().filter(m -> {
            return m.getTableCompares() != null && m.getTableCompares().stream().anyMatch(c -> ModifyType.DROP.equals(c.getModifyType()) && ModifyType.ModifyLevel.TABLE.equals(c.getModifyLevel()));
        }).collect(Collectors.toList());
        if (dropTableCrs.isEmpty()) {
            return r2;
        }
        // 找出r2的drop table中，在r1中为alter table的（说明需要修复该表的对比结果：把r1的放到r2，r2和该表相关的去掉）
        List<ModelCompareVo> alterTableR1 = new ArrayList<>();
        Set<ModelCompareVo> outCrR2 = new HashSet<>();
        for (ModelCompareVo _r1Mc : r1) {
            for (ModelCompareVo _r2DropTableMc : dropTableCrs) {
                if (_r1Mc.getTableId().equals(_r2DropTableMc.getTableId()) ||
                        _r1Mc.getTableName().equals(_r2DropTableMc.getTableName())) {
                    alterTableR1.add(_r1Mc);
                    outCrR2.add(_r2DropTableMc);
                }
            }
        }
        List<ModelCompareVo> res = new ArrayList<>();
        for (ModelCompareVo r : r2) {
            for (ModelCompareVo _r : outCrR2) {
                if (!r.getTableName().equals(_r.getTableName()) && !r.getTableId().equals(_r.getTableId())) {
                    res.add(r);
                }
            }
        }
        res.addAll(alterTableR1);
        return res;
    }

    /**
     * 是否强制进行清除上一次对比结果数据
     *
     * @param forceOverride 是否强制
     * @param notDoneTasks  上一次带同步任务
     */
    private void forceOverrideTasks(Boolean forceOverride, List<DapSyncProdEnvTask> notDoneTasks) {
        if (forceOverride) {
            notDoneTasks.forEach(t -> {
                t.setBmStatus(BmStatus.DELETE);
                boolean status = taskService.updateById(t);
                // 删除明细
                LambdaUpdateWrapper<DapSyncProdEnvTaskDetail> delDetails = Wrappers.<DapSyncProdEnvTaskDetail>lambdaUpdate()
                        .eq(DapSyncProdEnvTaskDetail::getTaskId, t.getId())
                        .set(DapSyncProdEnvTaskDetail::getBmStatus, BmStatus.DELETE);
                boolean update = taskDetailService.update(delDetails);
                log.warn("接口请求执行强制覆盖模型上一次对比结果，更新非已同步任务(包含任务明细表)为删除，taskId={}，status:{}, detailStatus:{}", t.getId(), status, update);
            });
            notDoneTasks.clear();
        }
    }

    /**
     * 异步保存任务和明细
     *
     * @param globalConfig 全局配置
     * @param compareR     对比结果
     * @return SyncProTaskDto
     */
    private SyncProTaskDto asyncSaveTaskAndDetails(GlobalConfig globalConfig, List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR) {
        DapDataModel model = globalConfig.getModel();
        DapSyncProdEnvTask task = getTask(model);
        List<DapSyncProdEnvTaskDetail> details = getTaskDetails(compareR, task);
        SyncProTaskDto syncProTaskDto = SyncProTaskDto.builder().task(task).taskDetails(details).build();
        log.info("对比结果：单号={}，变更数={}", syncProTaskDto.getTask().getChangeOrder(), syncProTaskDto.getTaskDetails().size());
        return syncProTaskDto;
    }

    /**
     * 获取任务
     *
     * @param model 模型
     * @return 任务主表
     */
    private DapSyncProdEnvTask getTask(DapDataModel model) {
        DapSyncProdEnvTask task = new DapSyncProdEnvTask();
        // 写入对比结果到发布同步任务中
        task.setChangeOrder("");
        task.setApplicationBy("");
        task.setVerifyBy("");
        task.setChangeDesc("");
        task.setApplicationTime(null);
        task.setModelId(model.getId());
        task.setDataSourceId(model.getDataSourceId());
        task.setPreProdStatus(SyncStatus.SYNC_WAIT);
        task.setProdStatus(SyncStatus.SYNC_WAIT);
        // 禁用状态，待发布通过后才修改为启用状态
        task.setBmStatus(BmStatus.DISABLE);
        task.setId(IdWorker.getIdStr());
        return task;
    }

    /**
     * 任务单号-表名明细ddl
     *
     * @param compareR 对比结果
     * @param task     任务主表
     * @return 明细
     */
    private List<DapSyncProdEnvTaskDetail> getTaskDetails(List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR, DapSyncProdEnvTask task) {
        List<DapSyncProdEnvTaskDetail> details = compareR.stream().map(tableInfo -> {
            List<CompareResult<SqlCompareResult>> compareResults = tableInfo.values().stream().flatMap(Collection::stream).collect(Collectors.toList());
            List<DapSyncProdEnvTaskDetail> taskDetails = compareResults.stream().map(c -> {
                SqlCompareResult result = c.getData();
                DapSyncProdEnvTaskDetail taskDetail = new DapSyncProdEnvTaskDetail();
                taskDetail.setId(IdWorker.getIdStr());
                taskDetail.setModelId(c.getModelId());
                taskDetail.setTableId(c.getTableId());
                taskDetail.setTableName(c.getTableName());
                taskDetail.setTaskId(task.getId());
                // 收集表级别的变更ddl
                if (c.getModifyLevel().equals(ModifyType.ModifyLevel.TABLE)) {
                    taskDetail.setPtDdl(result.getScript().getSql());
                }
                taskDetail.setDdl(result.getScript().getSql());
                taskDetail.setPtOsc(result.getScript().getPtSql());
                taskDetail.setPreProdScriptStatus(SyncStatus.SYNC_WAIT);
                taskDetail.setProdScriptStatus(SyncStatus.SYNC_WAIT);
                // 禁用状态，待发布通过后才修改为启用状态
                taskDetail.setBmStatus(BmStatus.DISABLE);
                return taskDetail;
            }).collect(Collectors.toList());
            // 对一个表多个变更的组合
            // 1)只有一个变更直接把ddl
            if (taskDetails.size() == 1) {
                return taskDetails.get(0);
            }
            // 2)多个变更合并
            DapSyncProdEnvTaskDetail taskDetail = taskDetails.stream().reduce((d1, d2) -> {
                // 组合可执行的sql+ptosc
                if (d2.getPtDdl() != null && d2.getPtOsc() == null) {
                    String ptDdl = Optional.ofNullable(d1.getPtDdl()).map(d -> concatAndLine(d, SQL_SPLIT_CHAR).concat(d2.getPtDdl())).orElse(d2.getPtDdl());
                    d1.setPtDdl(ptDdl);
                } else {
                    String ptOsc = Optional.ofNullable(d1.getPtOsc()).map(d -> concatAndLine(d, PT_SQL_SPLIT_CHAR).concat(d2.getPtOsc())).orElse(d2.getPtOsc());
                    d1.setPtOsc(ptOsc);
                }
                // 收集以上比较出来的d1#ddl结果
                String ddl = Optional.ofNullable(d1.getDdl()).map(d -> concatAndLine(d, SQL_SPLIT_CHAR).concat(d2.getDdl()))
                        .orElseGet(() -> {
                            List<String> res = Stream.of(d1.getDdl(), d2.getDdl()).filter(StringUtils::isNotBlank).collect(Collectors.toList());
                            return String.join(concatAndLine("", SQL_SPLIT_CHAR), res);
                        });
                d1.setDdl(ddl);
                return d1;
            }).orElseThrow(() -> new DapThrowException("合并表中多个变更记录失败！！"));
            return taskDetail;
        }).collect(Collectors.toList());
        return details;
    }

    /**
     * 申请发布通过-
     *
     * @param apply 发布内容
     */
    @Override
    public void applyConfirmPass(ApplyPublishParam apply, UserInfoHolder.UserInfo userInfo) {
        updateTaskStatusByApplyAction(apply, BmStatus.ON, userInfo);
    }

    @Override
    @Transactional(rollbackFor = {DapException.class, RuntimeException.class})
    public void updateTaskStatusByApplyAction(ApplyPublishParam apply, BmStatus bmStatus, UserInfoHolder.UserInfo userInfo) {
        BmAssetUtils.notNull(bmStatus, "指定同步任务的最终状态");
        String modelId = apply.getModelId();
        String compareVersion = apply.getCompareVersion();
        // 获取禁用状态的数据，进行更新状态和相关申请发布信息
        DapSyncProdEnvTask task = taskService.getByIdAsset(compareVersion, "入参对比版本号不存在");
        // 增加更新约束，只允许状态为禁用+预发布状态为未同步+发布状态为未同步
        BmAssetUtils.isTrue(BmStatus.DISABLE.equals(task.getBmStatus()) &&
                SyncStatus.SYNC_WAIT.equals(task.getPreProdStatus()) &&
                SyncStatus.SYNC_WAIT.equals(task.getProdStatus()), "版本号所对应任务必须是未同步状态，不可进行" + ((bmStatus.equals(BmStatus.DELETE) ? "撤销" : "通过")));
        task.setBmStatus(bmStatus);
        task.setChangeOrder(apply.getChangeOrder());
        task.setApplicationBy(apply.getApplicationBy());
        task.setVerifyBy(apply.getVerifyBy());
        task.setChangeDesc(apply.getChangeDesc());
        task.setApplicationTime(apply.getApplicationTime());
        boolean status = taskService.updateById(task);
        log.info("申请发布-确认{}，modelId={}, taskId={}, status={}, info={}", (bmStatus.equals(BmStatus.DELETE) ? "撤销" : BmStatus.DISABLE.equals(bmStatus) ? "不通过" : "通过"), modelId, compareVersion, status, apply);
        LambdaUpdateWrapper<DapSyncProdEnvTaskDetail> updateWrapper = Wrappers.<DapSyncProdEnvTaskDetail>lambdaUpdate().eq(DapSyncProdEnvTaskDetail::getModelId, modelId)
                .eq(DapSyncProdEnvTaskDetail::getTaskId, compareVersion)
                .eq(DapSyncProdEnvTaskDetail::getBmStatus, BmStatus.DISABLE)
                .set(DapSyncProdEnvTaskDetail::getBmStatus, bmStatus);
        boolean update = taskDetailService.update(updateWrapper);
        log.info("更新任务明细：任务为taskId = {},  status={}", compareVersion, update);
        String msg;
        if (BmStatus.ON.equals(bmStatus))
            msg = String.format("申请发布-%s-通过，进入发布同步列表，执行人：%s", apply.getAutoPassDesc(), userInfo.getUserName());
        else if (BmStatus.DELETE.equals(bmStatus))
            msg = String.format("申请发布-撤销，申请发布流程结束，执行人：%s", userInfo.getUserName());
        else
            msg = String.format("申请发布-不通过，申请发布流程回退，执行人：%s", userInfo.getUserName());
        taskOperRecordService.mark(modelId, compareVersion, msg);
    }

    /**
     * 申请发布 草稿撤销
     *
     * @param apply 发布内容
     */
    @Override
    public void applyUnPass(ApplyPublishParam apply, UserInfoHolder.UserInfo userInfo) {
        updateTaskStatusByApplyAction(apply, BmStatus.DELETE, userInfo);
    }

    @Override
    public void applyConfirmUnPass(ApplyPublishParam apply, UserInfoHolder.UserInfo userInfo) {
        updateTaskStatusByApplyAction(apply, BmStatus.DISABLE, userInfo);
    }

    /**
     * @param globalConfig 全局配置
     * @param mock         true 针对获取的两个版本可能不存在差异的情况下，直接构造所有属性都发生变化 ，false取真实数据
     * @return 结果集
     */
    private Tuple2<List<TableModel>, List<TableModel>> getV12Model(GlobalConfig globalConfig, boolean mock) {
        List<TableModel> modelV1 = getLocalTableModel(globalConfig, BmKeyType.USER_NAME_AND_PID, false, mock);
        List<TableModel> modelV2 = getLocalTableModel(globalConfig, BmKeyType.USER_NAME_AND_ID, true, mock);
        return Tuples.of(modelV1, modelV2);
    }

    /**
     * @param globalConfig 全局配置
     * @param mock         true 针对获取的两个版本可能不存在差异的情况下，直接构造所有属性都发生变化 ，false取真实数据
     * @return 结果集
     */
    private Tuple2<List<TableModel>, List<TableModel>> getV12ModelByIdOrPid(GlobalConfig globalConfig, boolean mock) {
        List<TableModel> modelV1 = getLocalTableModel(globalConfig, BmKeyType.USED_PID, false, mock);
        List<TableModel> modelV2 = getLocalTableModel(globalConfig, BmKeyType.USED_ID, true, mock);
        return Tuples.of(modelV1, modelV2);
    }

    /**
     * 获取本地模型版本数据
     *
     * @param globalConfig 全局配置
     * @param modelNewest  是否最新数据 true 最新模型数据 false最新版本数据
     * @param mockData     模拟版本数据的差异（测试对比用）
     * @return 版本数据
     */
    private List<TableModel> getLocalTableModel(GlobalConfig globalConfig, BmKeyType useIdBmKey, boolean modelNewest, boolean mockData) {
        globalConfig.setModelNewest(modelNewest);
        // 模拟模型v1数据
        List<TableModel> modelV1 = syncBasicService.getLocalTableModel(globalConfig)
                .stream()
                .peek(t -> {
                    DapDataModelTable table = t.getTable();
                    // v1拿pid对应到最新模型id
                    if (useIdBmKey.equals(BmKeyType.USER_NAME)) {
                        table.setBmKey(table.getEngName());
                        t.getFields().forEach(f -> f.setBmKey(f.getFieldName()));
                        t.getIndexes().forEach(i -> i.setBmKey(i.getKeyName()));
                        t.getPartitions().forEach(p -> p.setBmKey(p.getPartitionFieldName()));
                    }
                    if (useIdBmKey.equals(BmKeyType.USER_NAME_AND_ID)) {
                        table.setBmKey(table.getEngName());
                        t.getFields().forEach(f -> f.setBmKey(f.getId()));
                        t.getIndexes().forEach(i -> i.setBmKey(i.getId()));
                        t.getPartitions().forEach(p -> p.setBmKey(p.getId()));
                    }
                    if (useIdBmKey.equals(BmKeyType.USER_NAME_AND_PID)) {
                        table.setBmKey(table.getEngName());
                        t.getFields().forEach(f -> f.setBmKey(f.getPId()));
                        t.getIndexes().forEach(i -> i.setBmKey(i.getPId()));
                        t.getPartitions().forEach(p -> p.setBmKey(p.getPId()));
                    }
                    if (useIdBmKey.equals(BmKeyType.USED_ID)) {
                        table.setBmKey(table.getId());
                        t.getFields().forEach(f -> f.setBmKey(f.getId()));
                        t.getIndexes().forEach(i -> i.setBmKey(i.getId()));
                        t.getPartitions().forEach(p -> p.setBmKey(p.getId()));
                    }
                    if (useIdBmKey.equals(BmKeyType.USED_PID)) {
                        table.setBmKey(table.getPId());
                        t.getFields().forEach(f -> f.setBmKey(f.getPId()));
                        t.getIndexes().forEach(i -> i.setBmKey(i.getPId()));
                        t.getPartitions().forEach(p -> p.setBmKey(p.getPId()));
                    }
                })
                .peek(t -> {
                    // 模拟版本差异的生成
                    if (mockData) {
                        String suffix = IdWorker.getIdStr();
                        String name = t.getTable().getName() + "_" + suffix;
                        t.getTable().setName(name);
                        // 增加修改表英文名 当前对比基于id，不是表英文名
                        String engName = t.getTable().getEngName() + "_" + suffix;
                        t.getTable().setEngName(engName);
                        t.getFields().forEach(i -> {
                            i.setComment(i.getComment() + "_" + suffix);
                        });
                        t.getIndexes().forEach(i -> {
                            i.setKeyChinName(i.getKeyChinName() + "_" + suffix);
                        });
                    }
                })
                .collect(Collectors.toList());
        return modelV1;
    }

    public enum BmKeyType {
        USED_ID, USED_PID, USER_NAME, USER_NAME_AND_ID, USER_NAME_AND_PID
    }


}
